package cn._51doit.flink.day05;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * 先keyBy，然后按照Event划分滚动窗口
 * 底层调用的是window方法，返回的是keyedWindow，window和window operator对应的Task是多并行的
 *
 * 窗口触发后，每个分区中，每一个组的数据都会产生结果，然后输出
 *
 * 1.Flink中EventTime类型的窗口，是按照数据中的EventTime触发的，EventTime要转换成long类型的，精确到毫秒的时间戳
 * 2.窗口是根据输入的数据中的EventTime确定的，窗口的起始时间、结束时间是对齐的，是窗口长度的整数倍，而且是前闭后开的 [1645252200000, 1645252205000)
 *
 *
 */
public class EventTimeTumblingWindowDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //1645252201000,spark,4
        //1645252202000,hive,5
        //1645252204000,hive,2
        //1645252204998,spark,1
        //1645252204999,spark,2
        //1645252206666,flink,2
        //1645252207777,spark,3
        //1645252208888,spark,1
        //1645252209998,spark,2
        //1645252210000,spark,200
        //1645252214999,spark,100
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //提取数据中的EventTime，按照数据中的时间划分窗口
        //该方法仅是提取数据中的时间，不会改变原有数据的样子
        //WaterMark是一种特殊的消息，这种特殊的小时是提取EventTime的算子，想下游发送的，发送给窗口对应的Task
        //WaterMark = 每个分区中最大的EventTime - 延迟时间
        //窗口触发的时机为：WaterMark >= 窗口的结束时间，窗口触发
        SingleOutputStreamOperator<String> linesWithWaterMark = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]); //将数据中的数据提取出来，返回long类型的时间戳
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = linesWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String line) throws Exception {
                String[] fields = line.split(",");
                return Tuple2.of(fields[1], Integer.parseInt(fields[2]));
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAndCount.keyBy(t -> t.f0);
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)));
        SingleOutputStreamOperator<Tuple2<String, Integer>> res = windowedStream.sum(1);
        res.print();
        env.execute();


    }
}
